package net.wsnetwork.jawa.frontend;

import java.io.IOException;
import java.io.OutputStream;
import java.util.logging.Logger;

import net.wsnetwork.jawa.common.ChunkedOutputStream;
import net.wsnetwork.jawa.common.ConnectionProcessingException;
import net.wsnetwork.jawa.common.Constants;
import net.wsnetwork.jawa.common.HttpConnectionHandler;

import org.apache.commons.io.IOUtils;
import org.apache.http.Header;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpUriRequest;
import org.apache.http.client.params.AllClientPNames;
import org.apache.http.client.protocol.RequestAddCookies;
import org.apache.http.client.protocol.ResponseProcessCookies;
import org.apache.http.conn.ConnectTimeoutException;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.http.impl.conn.tsccm.ThreadSafeClientConnManager;
import org.apache.http.message.HeaderGroup;
import org.apache.http.protocol.RequestTargetHost;
import org.apache.http.protocol.RequestUserAgent;

/**
 * An extremely simple HTTP server implementation that passes through requests to the backend
 * cache servers. Using the Apache HTTP client, requests are bundled together using HTTP keep-alive
 * to decrease the number of connections made to the backend.
 * 
 * @author cairns@colorado.edu (Brian Cairns)
 */
public class FrontendConnectionHandler extends HttpConnectionHandler {
    private static final Logger LOG =
        Logger.getLogger(FrontendConnectionHandler.class.getCanonicalName());
    private static final byte[] CHUNKED_ENCODING =
        "Transfer-Encoding: chunked\r\n".getBytes(Constants.UTF_8);

    private final int RETRIES = 1;
    private final int MAX_BACKEND_CONNECTIONS = 10;
    private final DefaultHttpClient httpClient;
    private final CacheRouter router;

    public FrontendConnectionHandler(CacheRouter router) {
        ThreadSafeClientConnManager connManager = new ThreadSafeClientConnManager();
        connManager.setDefaultMaxPerRoute(MAX_BACKEND_CONNECTIONS);
        connManager.setMaxTotal(MAX_BACKEND_CONNECTIONS);
        httpClient = new DefaultHttpClient(connManager);
        httpClient.getParams().setIntParameter(AllClientPNames.CONNECTION_TIMEOUT, 10000);
        httpClient.getParams().setIntParameter(AllClientPNames.SO_TIMEOUT, 10000);
        httpClient.getParams().setBooleanParameter(AllClientPNames.HANDLE_REDIRECTS, false);
        httpClient.getParams().setBooleanParameter(AllClientPNames.HANDLE_AUTHENTICATION, false);
        httpClient.removeRequestInterceptorByClass(RequestAddCookies.class);
        httpClient.removeRequestInterceptorByClass(RequestTargetHost.class);
        httpClient.removeRequestInterceptorByClass(RequestUserAgent.class);
        httpClient.removeResponseInterceptorByClass(ResponseProcessCookies.class);
        this.router = router;
    }

    @Override
    protected boolean handleHttpConnection(OutputStream out, String uri, HeaderGroup headers)
    throws IOException, ConnectionProcessingException {
        Header hostHeader = headers.getFirstHeader("host");
        String host = hostHeader.getValue();
        boolean keepAlive = isKeepalive(headers);

        HttpResponse response = null;
        for (int i = 0; i < RETRIES; ++i) {
            String urlString = "http://" + router.getBackendForRequest(host, uri).getHostAddress() +
            ":8080" + uri;
            HttpUriRequest get = new HttpGet(urlString);

            LOG.info("Request to backend: " + urlString);

            // Connection to the back-end cache server is always keep-alive
            get.setHeader("Connection", "keep-alive");

            // Pass through request headers
            for (Header header : headers.getAllHeaders()) {
                if (isHeaderAllowed(header)) {
                    get.setHeader(header.getName(), header.getValue());
                }
            }

            // Make request
            try {
                response = httpClient.execute(get);
            } catch (ConnectTimeoutException e) {
                get.abort();
                LOG.info("Timed out: " + urlString);
                // Unable to contact back-end; try again
                break;
            }
        }

        if (response == null) {
            String responseText = "HTTP/1.1 500 Too Busy\r\n\r\n";
            out.write(responseText.getBytes(Constants.UTF_8));
            return false;
        }

        // Copy response status line
        String statusLine = response.getStatusLine().getProtocolVersion() + " "
        + response.getStatusLine().getStatusCode() + " "
        + response.getStatusLine().getReasonPhrase() + "\r\n";
        out.write(statusLine.getBytes(Constants.UTF_8));

        // Copy response headers
        for (Header header : response.getAllHeaders()) {
            String headerString = header.getName() + ": " + header.getValue() + "\r\n";
            if (isHeaderAllowed(header)) {
                out.write(headerString.getBytes(Constants.UTF_8));
            }
        }

        // If keep-alive and no response length, we need to use chunked encoding
        ChunkedOutputStream chunkedOut = null;
        if (keepAlive && response.getEntity() != null &&
                response.getFirstHeader("content-length") == null) {
            out.write(CHUNKED_ENCODING);
            chunkedOut = new ChunkedOutputStream(out);
        }

        // Send response connection header
        String connectionType = keepAlive ? "keep-alive" : "close";
        String connectionHeaderString = "Connection: " + connectionType + "\r\n";
        out.write(connectionHeaderString.getBytes(Constants.UTF_8));

        // End headers
        out.write("\r\n".getBytes(Constants.UTF_8));

        // Return response content, if any
        if (response.getEntity() != null) {
            if (chunkedOut != null) {
                IOUtils.copy(response.getEntity().getContent(), chunkedOut);
            } else {
                IOUtils.copy(response.getEntity().getContent(), out);
            }
        }
        if (chunkedOut != null) {
            chunkedOut.finish();
        }
        out.flush();
        response.getEntity().getContent().close();

        return keepAlive;
    }

    private static boolean isKeepalive(HeaderGroup headers) {
        Header requestConnectionHeader = headers.getFirstHeader("connection");
        return (requestConnectionHeader != null &&
                requestConnectionHeader.getValue().toLowerCase().equals("keep-alive"));
    }

    /**
     * Returns {@code true} if the given request header should be passed through in the request.
     */
    private static boolean isHeaderAllowed(Header header) {
        String name = header.getName().toLowerCase();
        if (name.equals("connection")) { return false; }
        if (name.equals("transfer-encoding")) { return false; }
        return true;
    }
}
